/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.queue.impl;

import java.lang.reflect.InvocationTargetException;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;

import org.sabayframework.activity.Activity;
import org.sabayframework.activity.ActivityContext;
import org.sabayframework.activity.ActivityMonitor;
import org.sabayframework.channel.Channel;
import org.sabayframework.channel.Lock;
import org.sabayframework.queue.ExecutionQueue;

/**
 * Implementation of ActivityContext interface.
 * 
 * @author Alex Antonau
 * */
public final class ActivityContextImpl<I, O> implements ActivityContext<I, O>, ActivityMonitor<I,O> {

	private ExecutionQueue queue;
	private final Activity<I, O> activity;
	private Channel<I> input;
	private Channel<O> output;
	private final String activityName;
	
	private final AtomicLong executionTime;
	private final AtomicLong callCount;
	private final long startTime;
	private long stopTime;
	private volatile boolean alive;
	private final ActivityMonitor<?, ?> parent;
	private volatile Throwable error;

	public ActivityContextImpl(ExecutionQueue queue,
			Activity<I, O> activity, Channel<I> input,
			Channel<O> output, ActivityMonitor<?, ?> parent) {
		super();
		this.queue = queue;
		this.activity = activity;
		this.input = input;
		this.output = output;
		this.alive = true;
		
		this.executionTime = new AtomicLong();
		this.callCount = new AtomicLong();
		this.activityName = activity.toString();
		this.startTime = System.currentTimeMillis();
		this.parent = parent;
	}

	@Override
	public long getLiveTime() {
		return alive? System.currentTimeMillis() - startTime:
				stopTime - startTime;
	}

	@Override
	public boolean isAlive() {
		return alive;
	}

	public String getActivityName() {
		return activityName;
	}

	@Override
	public Channel<I> getInput() {
		return input;
	}

	@Override
	public Channel<O> getOutput() {
		return output;
	}

	@Override
	public Lock<I> getReadLock(final int size) {
		return input.getReadLock(size);
	}

	@Override
	public Lock<O> getWriteLock(final int size) {
		return output.getWriteLock(size);
	}

	@Override
	public ExecutionQueue getExecutionQueue() {
		return queue;
	}
	@Override
	public Activity<I, O> getActivity() {
		return activity;
	}

	@Override
	public long getExecutionTime() {
		return executionTime.get();
	}

	@Override
	public long getCallCount() {
		return callCount.get();
	}
	
	public void handleError(Throwable t) {
		input.fail(t);
		output.fail(t);
	}

	public boolean isReady() {
		try {
			return activity.isReady(this);
		} catch (Throwable e) {
			handleError(e);
		}
		return false;
	}

	public boolean isDone() {
		try {
			return activity.isDone(this);
		} catch (Throwable e) {
			handleError(e);
		}
		return true;
	}

	public void beforeAdding(ExecutionQueue queue) {
		try {
			activity.beforeExecute(this);
		} catch (Throwable e) {
			handleError(e);
		}
	}

	public void afterRemoving(ExecutionQueue queue) {
		try {
			activity.afterExecute(this);
		} catch (Throwable e) {
			handleError(e);
		}
		stopTime = System.currentTimeMillis();
		alive = false;
	}
	
	/**
	 * Execute one cycle of the activity on the input queue.
	 * @return execution time, ms
	 * @throws Exception 
	 * */
	long runOnce() throws Throwable{
		if(error != null) throw error;
		final long startTime = System.currentTimeMillis();
		long delta = 0; 
		try{
			callCount.incrementAndGet();
			activity.execute(this);
		}
		finally{
			delta = System.currentTimeMillis()-startTime;
			executionTime.addAndGet(delta);
		}
		return delta;
	}

	@Override
	public void kill(Throwable error) {
		handleError(error);
		this.error = error;
	}

	@Override
	public O waitForResult(long timeoutMillis)
			throws InvocationTargetException, InterruptedException,
			TimeoutException {
		Lock<O> lock = output.waitForReadLock(1, timeoutMillis);
		if(lock!=null){
			try{
				return lock.get(0);
			} finally{
				lock.unlock();
			}
		}
		if(output.hasFailed()){
			throw new InvocationTargetException(output.getFailureCause());
		}
		if(output.isClosed()){
			throw new IllegalStateException();
		}
		throw new TimeoutException(String.valueOf(timeoutMillis));
	}

	@Override
	public ActivityMonitor<?, ?> getParent() {
		return parent;
	}

	@Override
	public int hashCode() {
		return activity.hashCode();
	}

}
